import org.apache.spark.{SparkConf, SparkContext}

object PeopleInfoCalc {
  def main(args: Array[String]): Unit = {
    //����SparkContext
    val conf = new SparkConf().setAppName("PeopleInfoCalc").setMaster("local")
    //    val conf = new SparkConf().setAppName("PeopleInfoCalc")
    val sc = new SparkContext(conf)

    //��ȡ����
    val dataFile = sc.textFile("/Users/zhangyimin/training/data/input/sample_people_info.txt")
    //    val dataFile = sc.textFile(args(0))

    //��֣��С�Ů                                                                                                     ����     2 M 105             �Ա�                                                      ����
    val maleData = dataFile.filter(line => line.contains("M")).map(line => (line.split(" ")(1) + " " + line.split(" ")(2)))
    val feMaleData = dataFile.filter(line => line.contains("F")).map(line => (line.split(" ")(1) + " " + line.split(" ")(2)))
    for (x<-maleData)print(x)

    //�õ����Ե����
    val maleHeighData = maleData.map(line => line.split(" ")(1).toInt)
    val femaleHeighData = feMaleData.map(line => line.split(" ")(1).toInt)

//    println(maleHeighData.collect + "和" + femaleHeighData.collect)

    //使用sortBy 排序 求最低
    //    val lowestMale = maleHeighData.sortBy(x=>x, true).first()
    //    val lowestFeMale = femaleHeighData.sortBy(x=>x, true).first()


    //使用agreegate求最高身高
    val highesttMale = maleHeighData.aggregate(0)(math.max(_, _), math.max(_, _))
    val highestFeMale = femaleHeighData.aggregate(0)(math.max(_, _), math.max(_, _))


    //使用sortBy 排序 求最高
    //    val highesttMale = maleHeighData.sortBy(x=>x, false).first()
    //    val highestFeMale = femaleHeighData.sortBy(x=>x, false).first()

    //使用agreegate求最低身高
    val lowestMale = maleHeighData.aggregate(1000)(math.min(_, _), math.min(_, _))
    val lowestFeMale = femaleHeighData.aggregate(1000)(math.min(_, _), math.min(_, _))

    //���
    println("lowestMale:" + lowestMale)
    println("lowestFeMale:" + lowestFeMale)
    println("highesttMale:" + highesttMale)
    println("highestFeMale:" + highestFeMale)

    sc.stop()
  }
}